package cn.aijson.demo.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Author iqlife
 * Desc 使用SparkStreaming接收node1:9999的数据并做WordCount
 * 实现状态管理
 */
object WordCountStreaming {
  def creatingFunc():StreamingContext ={
    //1、准备spark context
    var conf:SparkConf=new SparkConf().setAppName("workdcount").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    //设置流，//每隔5s划分一个批次
    val ssc: StreamingContext = new StreamingContext(sc, Seconds(5))

    //注意:state存在checkpoint中
    ssc.checkpoint("data/checkpoint")

    //2、加载数据
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    //3、处理数据-直接累加，5秒之后重新开始
    val resultRDD: DStream[(String, Int)] = lines.flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)

    //checkpoint 存储处理数据
    //定义一个函数用来处理状态:把当前数据和历史状态进行累加
    //currentValues:表示该key(如:spark)的当前批次的值,如:[1,1]
    //historyValue:表示该key(如:spark)的历史值,第一次是0,后面就是之前的累加值如1
    val updateFunc = (currentValues: Seq[Int], historyValue: Option[Int]) => {
      if (currentValues.size > 0) {
        val currentResult: Int = currentValues.sum + historyValue.getOrElse(0)
        Some(currentResult)
      } else {
        historyValue
      }
    }

    //4、处理数据-5秒的计算结果存入checkpoint
    val checkpointRDD: DStream[(String, Int)] = lines.flatMap(_.split(" "))
      .map((_, 1))
      .updateStateByKey(updateFunc)

    //5、输出结果
    resultRDD.print()
    checkpointRDD.print()
    ssc
  }

  def main(args: Array[String]): Unit = {

    //1、判断环境是否已经存在checkpoint ，如果存在，则先恢复checkpoint
    val ssc: StreamingContext = StreamingContext.getOrCreate("data/checkpoint", creatingFunc _)
    ssc.sparkContext.setLogLevel("WARN")

    //2.启动并等待结束
    ssc.start()
    ssc.awaitTermination()//注意:流式应用程序启动之后需要一直运行等待手动停止/等待数据到来

    //3.关闭资源
    ssc.stop(stopSparkContext = true, stopGracefully = true)//优雅关闭
  }

}
